﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Net.Sockets;
using System.Net;
using System.IO;
using System.Net.Security;
using PServiceBus.Core.Interface;
using System.Collections;
using PServiceBus.Core.Runtime.Extensions;
using PServiceBus.Core.Runtime;
using PServiceBus.Core.Logger;
using System.Text.RegularExpressions;
using System.Threading.Tasks;
using System.Collections.Concurrent;
using System.Threading;

namespace PServiceBus.Tcp {
    public sealed class Provider<TObject> : IObjectProvider<TObject> where TObject : class {
        
        private TcpListener _listener;
        private const int LENGTH_BUFFER_SIZE = 4;
        private const int BUFFER_SIZE = (1024 * 1024) * 10;
        private const int SEND_BUFFER_RETRY_COUNT = 5;
        private const int SEND_BUFFER_SIZE = 1048576 * 10;
        private const int RECEIVE_BUFFER_SIZE = 1048576 * 10;
        private const int MAX_TCP_CONNECTION = 20000;
        private static ConcurrentDictionary<string, TcpClient> _clients = new ConcurrentDictionary<string, TcpClient>();
        private static ConcurrentDictionary<string, object> _providers = new ConcurrentDictionary<string, object>();
        private BlockingCollection<byte[]> _dataQueue = new BlockingCollection<byte[]>();
        private string _ipAddress;
        private volatile bool _running;
        private int _port;
        private bool _useSSL;


        public static Provider<TObject> GetProvider(string endpoint){
            return _providers.GetOrAdd(endpoint, key => new Provider<TObject> { Endpoint = key }) as Provider<TObject>;
        }

        private TcpClient GetClient(string key, string host, int port) {
            return _clients.GetOrAdd(key, k => NewClient(host, port));
        }

        private TcpClient Client { get { return GetClient(Endpoint, Host, Port); } }

        private static TcpClient NewClient(string host, int port) {
            var client = new TcpClient();
            client.Client.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.SendBuffer, SEND_BUFFER_SIZE);
            client.Client.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReceiveBuffer, RECEIVE_BUFFER_SIZE);
            client.LingerState = new LingerOption(false, 0);
            client.NoDelay = true;
            client.Connect(new IPEndPoint(IPAddress.Parse(host), port));
            return client;
        }

        
        public string Host {
            get {
                ParseEndPoint();
                return _ipAddress;
            }
        }

        public int Port {
            get {
                ParseEndPoint();
                return _port;
            }
        }

        public bool UseSSL {
            get {
                ParseEndPoint();
                return _useSSL;
            }
        }

        private void Send(TObject message) {
            Send(message.Serialize());
        }

        private void Send(byte[] data, bool retry = false, int retryCounter = 0) {
            try {
                if (!retry) {
                    var lengthBuffer = BitConverter.GetBytes(data.Length);
                    if (BitConverter.IsLittleEndian) Array.Reverse(lengthBuffer);
                    var bufferList = new List<byte>();
                    bufferList.AddRange(lengthBuffer); bufferList.AddRange(data);
                    data = bufferList.ToArray();
                }
                Client.Client.Send(data);
            } catch(SocketException ex) {
                if ((ex.SocketErrorCode == SocketError.ConnectionReset ||
                    ex.SocketErrorCode == SocketError.NotConnected) && retryCounter < SEND_BUFFER_RETRY_COUNT) {
                    var tmpClient = default(TcpClient);
                    _clients.TryRemove(Endpoint, out tmpClient);
                    Send(data, retry: true, retryCounter: retryCounter + 1);
                } else throw;
            } catch { 
                throw;
            }
        }

        private void ParseEndPoint() {
            if (_ipAddress != null) return;
            if (String.IsNullOrWhiteSpace(Endpoint)) 
                throw new Exception("Endpoint for provider cannot be null or empty");
            var tokens = Endpoint.Split(':');
            var address = Dns.GetHostAddresses(tokens[0])
                        .FirstOrDefault(x => x.AddressFamily == System.Net.Sockets.AddressFamily.InterNetwork);
            _ipAddress = address != null ? address.ToString() : tokens[0];
            _port = Convert.ToInt32(tokens[1]);
            _useSSL = tokens.Length > 2 ? Convert.ToBoolean(tokens[2]) : false;
        }

        private void InitListener() {
            if (_listener != null) return;
            _listener = new TcpListener(new IPEndPoint(IPAddress.Any, Convert.ToInt32(Endpoint)));
            _listener.Server.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.SendBuffer, SEND_BUFFER_SIZE);
            _listener.Server.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReceiveBuffer, RECEIVE_BUFFER_SIZE);
            _listener.Server.LingerState = new LingerOption(false, 0);
            _listener.Server.NoDelay = true;
            _running = true;
            _listener.Start(MAX_TCP_CONNECTION);
            _listener.BeginAcceptTcpClient(HandleAcceptClient, null);
        }

        #region IDisposable Members

        public void Dispose() {
            if (_listener != null) MethodHelper.Try(_listener.Stop);
        }

        #endregion

        #region IObjectProvider<TObject> Members

        public bool Refresh(TObject obj) {
            throw new NotImplementedException();
        }

        public bool Exists(TObject obj) {
            throw new NotImplementedException();
        }

        public TObject Get(Guid id) {
            throw new NotImplementedException();
        }

        public bool Delete(TObject obj) {
            throw new NotImplementedException();
        }

        public bool Add(IEnumerable<TObject> list) {
            return MethodHelper.Try(() =>
            {
                foreach (var obj in list) Send(obj);
            }) == null;
        }

        public bool Add(TObject obj) {
            Send(obj);
            return true;
        }

        public void Clear() {
            throw new NotImplementedException();
        }

        public void Delete() {
            throw new NotImplementedException();
        }

        public bool Delete(IEnumerable<TObject> list) {
            throw new NotImplementedException();
        }

        public bool Delete(IEnumerable<Func<TObject, bool>> conditions) {
            throw new NotImplementedException();
        }

        public bool Delete(Func<TObject, bool> condition) {
            throw new NotImplementedException();
        }

        public bool Exists(Func<TObject, bool> condition) {
            throw new NotImplementedException();
        }

        public TObject Get(Func<TObject, bool> condition) {
            throw new NotImplementedException();
        }

        public string Endpoint { get; set; }

        public string EndpointAlias { get; set; }

        public bool AppendTypeNameToEndpoint { get; set; }

        public bool DeleteObjectOnRead { get; set; }

        public Func<TObject, Guid> IDFunc { get; set; }
        #endregion


        private class MessageFrame {
            public MemoryStream Buffer { get; set; }
            public Socket Client { get; set; }
            public byte[] ReceiveBuffer { get; set; }
            public byte[] LengthBuffer { get; set; }
            public int Length { get; set; }

            public void Reset() {
                Length = 0;
                Buffer.SetLength(0);
            }

            public static MessageFrame New(Socket client) {
                return new MessageFrame() { 
                    Client = client,
                    Buffer = new MemoryStream(),
                    LengthBuffer = new byte[LENGTH_BUFFER_SIZE],
                    ReceiveBuffer = new byte[BUFFER_SIZE]
                };
            }
        }

        private void HandleAcceptClient(IAsyncResult result) {
            if (!_running) return;
            try {
                var client = _listener.EndAcceptTcpClient(result);
                var socket = client.Client;

                var frame = MessageFrame.New(socket);

                socket.BeginReceive(frame.LengthBuffer, 0, LENGTH_BUFFER_SIZE, SocketFlags.None, HandleClient, frame);

                _listener.BeginAcceptTcpClient(HandleAcceptClient, null);
            } catch { }
        }

        private void ProcessReceive(MessageFrame frame, int startIndex, int reads) {
            var needed = (int)(frame.Length - frame.Buffer.Length);
            if (reads < needed)
                frame.Buffer.Write(frame.ReceiveBuffer, startIndex, reads);
            else {
                var remains = reads - needed;
                frame.Buffer.Write(frame.ReceiveBuffer, startIndex, needed);

                _dataQueue.TryAdd(frame.Buffer.ToArray());
                frame.Reset();

                startIndex += needed;

                ProcessRemain(frame, startIndex, remains);
            }
        }

        private void ProcessRemain(MessageFrame frame, int startIndex, int remains) {
            if (remains == 0) return;

            Buffer.BlockCopy(frame.ReceiveBuffer, startIndex, frame.LengthBuffer, 0, LENGTH_BUFFER_SIZE);
            if (BitConverter.IsLittleEndian) Array.Reverse(frame.LengthBuffer);
            frame.Length = BitConverter.ToInt32(frame.LengthBuffer, 0);
            
            startIndex += LENGTH_BUFFER_SIZE;
            remains -= LENGTH_BUFFER_SIZE;

            if (!(remains > 0)) return;

            ProcessReceive(frame, startIndex, remains);
        }

        private void HandleClient(IAsyncResult result) {
            var frame = result.AsyncState as MessageFrame;
            var socket = frame.Client;

            var reads = 0;

            try {
                reads = socket.EndReceive(result);
            } catch { }

            if (reads == 0) return;

            if (frame.Length == 0) {
                if (BitConverter.IsLittleEndian) Array.Reverse(frame.LengthBuffer);
                frame.Length = BitConverter.ToInt32(frame.LengthBuffer, 0);
            } else {
                ProcessReceive(frame, 0, reads);
            }

            var hasLength = frame.Length > 0;

            socket.BeginReceive(hasLength ? frame.ReceiveBuffer : frame.LengthBuffer,
                0, hasLength ? BUFFER_SIZE : LENGTH_BUFFER_SIZE,
                SocketFlags.None, HandleClient, frame);
        }

        #region IEnumerable<TObject> Members
        public IEnumerator<TObject> GetEnumerator() {
            InitListener();
            foreach (var data in _dataQueue.GetConsumingEnumerable())
                yield return data.Deserialize<TObject>();
        }
        #endregion

        #region IEnumerable Members

        IEnumerator IEnumerable.GetEnumerator() {
            return GetEnumerator();
        }

        #endregion
    }
}
